#ifndef __M_BROKER_H__
#define __M_BROKER_H__
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_threadpool.hpp"
#include "mq_host.hpp"
#include "mq_consumer.hpp"
#include "mq_connection.hpp"
namespace mymq
{
    #define HOST_NAME "MyVirtualHost"
    #define DBFILE "/meta.db"
    class Server
    {
    public:
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;

        Server(int port, const std::string& basedir)
        : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server",
                muduo::net::TcpServer::kReusePort),
        _dispatcher(std::bind(&Server::onUnknownMessage, this, std::placeholders::_1,
                std::placeholders::_2, std::placeholders::_3)),
        _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
        _host(std::make_shared<VirtualHost>(HOST_NAME, basedir, basedir + DBFILE)),
        _consumer_manager(std::make_shared<ConsumerManager>()),
        _connection_manager(std::make_shared<ConnectionManager>()),
        _threadpool(std::make_shared<ThreadPool>())
        {
            // 注册事件处理函数
            // 信道
            _dispatcher.registerMessageCallback<mymq::openChannelRequest>(std::bind(&Server::onOpenChannel, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mymq::closeChannelRequest>(std::bind(&Server::onCloseChanne, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 交换机
            _dispatcher.registerMessageCallback<mymq::declareExchangeRequest>(std::bind(&Server::onDeclareExchange, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mymq::deleteExchangeRequest>(std::bind(&Server::onDeleteExchange, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 队列
            _dispatcher.registerMessageCallback<mymq::declareQueueRequest>(std::bind(&Server::onDeclareQueue, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mymq::deleteQueueRequest>(std::bind(&Server::onDeleteQueue, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 绑定
            _dispatcher.registerMessageCallback<mymq::queueBindRequest>(std::bind(&Server::onQueueBind, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mymq::queueUnBindRequest>(std::bind(&Server::onQueueUnBind, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 消息发布
            _dispatcher.registerMessageCallback<mymq::basicPublishRequest>(std::bind(&Server::onBasicPublish, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mymq::basicAckRequest>(std::bind(&Server::onBasicAck, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 订阅队列消息
            _dispatcher.registerMessageCallback<mymq::basicConsumeRequest>(std::bind(&Server::onBasicConsume, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<mymq::basicCancelRequest>(std::bind(&Server::onBasicCancel, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 告知服务器：遇到什么事件调用什么函数
            _server.setConnectionCallback(
                std::bind(&Server::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        }

        void start()
        {
            _server.start();
            _baseloop.loop();
        }

    private:
        // 打开信道
        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& message, 
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("打开信道失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            myconn->openChannel(message);
        }
        // 关闭信道
        void onCloseChanne(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& message, 
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("关闭信道失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            myconn->closeChannel(message);
        }

        // 声明交换机
        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& message, 
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("声明交换机失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("声明交换机失败：没有找到信道");
                return;
            }
            cp->declareExchange(message);
        }

        // 删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& message, 
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("删除交换机失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("删除交换机失败：没有找到信道");
                return;
            }
            cp->deleteExchange(message);
        }

        // 声明队列
        void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& message, 
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("声明队列失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("声明队列失败：没有找到信道");
                return;
            }
            cp->declareQueue(message);
        }

        // 删除队列
        void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& message, 
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("删除队列失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("删除队列失败：没有找到信道");
                return;
            }
            cp->deleteQueue(message);
        }

        // 队列绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& message, 
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("队列绑定失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列绑定失败：没有找到信道");
                return;
            }
            cp->bind(message);
        }

        // 队列解除绑定
        void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn, const queueUnBindRequestPtr& message,
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("队列解除绑定失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列解除绑定失败：没有找到信道");
                return;
            }
            cp->unBind(message);
        }

        // 消息发布
        void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& message,
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("消息发布失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("消息发布失败：没有找到信道");
                return;
            }
            cp->basicPublish(message);
        }

        // 消息确认
        void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& message,
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("消息确认失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("消息确认失败：没有找到信道");
                return;
            }
            cp->basicAck(message);
        }

        // 队列消息订阅
        void onBasicConsume(const muduo::net::TcpConnectionPtr& conn, const basicConsumeRequestPtr& message,
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("队列消息订阅失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列消息订阅失败：没有找到信道");
                return;
            }
            cp->basicConsume(message);
        }

        // 队列消息取消订阅
        void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& message,
            muduo::Timestamp)
        {
            // 获取连接
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("队列消息取消订阅失败：没有找到连接对应的 Connection 对象");
                conn->shutdown();
                return;
            }
            // 获取信道
            Channel::ptr cp = myconn->getChannel(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("队列消息取消订阅失败：没有找到信道");
                return;
            }
            cp->basicCancel(message);
        }


        // 未知事件处理函数
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,
                            const MessagePtr& message,
                            muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

        void onConnection(const muduo::net::TcpConnectionPtr &conn) 
        {
            if (conn->connected())
            {
                _connection_manager->newConnection(_host, _consumer_manager, _codec, conn, _threadpool);
            }
            else
            {
                _connection_manager->delConnection(conn);
            }
        }

    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server; // 服务器对象
        ProtobufDispatcher _dispatcher; // 请求分发器，建立 请求-处理函数 的映射关系
        ProtobufCodecPtr _codec;        // 协议处理器，对收到的请求进行 protobuf 协议处理，得到结构化请求对象
        VirtualHost::ptr _host;
        ConsumerManager::ptr _consumer_manager;
        ConnectionManager::ptr _connection_manager;
        ThreadPool::ptr _threadpool;
    };
}

#endif